package io.github.bigbird0101.datatransfer.container;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import integration.data.task.api.channel.impl.MemoryNonBlockingChannel;
import integration.data.task.api.exchanger.impl.RecordExchanger;
import integration.data.task.api.transformation.impl.TransformerExecution;
import integration.data.task.api.utils.TransformerUtil;
import integration.data.task.reader.Reader;
import integration.data.task.writer.Writer;
import integration.modle.exception.IntegrationValidException;
import integration.utils.trace.LogUtils;
import io.github.bigbird0101.datatransfer.collector.Communication;
import io.github.bigbird0101.datatransfer.collector.CommunicationTool;
import io.github.bigbird0101.datatransfer.collector.StdoutPluginCollector;
import io.github.bigbird0101.datatransfer.collector.TaskPluginCollector;
import io.github.bigbird0101.datatransfer.constants.PluginType;
import io.github.bigbird0101.datatransfer.model.config.TransformerConfig;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.io.Serializable;
import java.util.List;

import static integration.data.task.api.constant.Constant.*;

/**
 * 逐条数据读写
 * 非阻塞任务容器
 */
@Slf4j
@Setter
@Getter
public class StreamCommonTaskContainer extends AbstractContainer implements Serializable {


    private String taskId;

    private String config;

    private long startTimeStamp;

    /**
     * 输入端
     */
    private String readerName;

    private String readerConfig;

    private Reader reader;

    /**
     * 转换
     */
    private String transformer;

    /**
     * 输出端
     */
    private String writerName;

    private String writerConfig;

    private Writer writer;

    /**
     * 限速配置
     */
    private String channelConfig;

    private RecordExchanger recordReceiver;

    private Communication lastTaskGroupContainerCommunication;

    /**
     * 该处的taskCommunication在多处用到：
     * 1. channel
     * 2. reader和writer的taskPluginCollector
     */
    private Communication taskCommunication;

    public StreamCommonTaskContainer() {
    }

    public StreamCommonTaskContainer(String config) {
        super(config);
        if (StrUtil.isBlank(config)) {
            throw new IntegrationValidException("数据同步配置参数不能为空");
        }
        this.config = config;
        JSONObject configJson = JSONUtil.toBean(config, JSONObject.class);
        readerName = configJson.getByPath(READER_NAME, String.class);
        writerName = configJson.getByPath(WRITER_NAME, String.class);
        taskId = configJson.getStr(TASK_ID);

        Object readerObj = configJson.getByPath(READER_PARAMETER);
        Object writerObj = configJson.getByPath(WRITER_PARAMETER);

        Object transformation = configJson.get(TRANSFORMATION);

        Object channelSetting = configJson.getByPath(CHANNEL_SETTING);

        if (writerName == null) {
            throw new IntegrationValidException("目标端类型为空");
        }
        if (readerObj == null) {
            throw new IntegrationValidException("来源端配置为空");
        }
        if (writerObj == null) {
            throw new IntegrationValidException("目标端配置为空");
        }
        readerConfig = JSONUtil.toJsonStr(readerObj);
        writerConfig = JSONUtil.toJsonStr(writerObj);

        if (transformation != null) {
            this.transformer = JSONUtil.toJsonStr(transformation);
        } else {
            this.transformer = EMPTY_ARRAY_JSON;
        }

        if (channelSetting != null) {
            this.channelConfig = JSONUtil.toJsonStr(channelSetting);
        } else {
            this.channelConfig = EMPTY_OBJECT_JSON;
        }


        this.taskCommunication = new Communication();

        LogUtils.info(log, "taskId:{}", taskId);
        LogUtils.info(log, "task config reader plugin: {}, reader config: {}", readerName, readerConfig);
        LogUtils.info(log, "task config writer plugin: {}, writer config: {}", writerName, writerConfig);
        LogUtils.info(log, "task config transformer config: {}", this.transformer);

        this.init();
    }

    public static StreamCommonTaskContainer getStreamCommonTaskContainer(String config) {
        return new StreamCommonTaskContainer(config);
    }

    public void init() {
        LogUtils.info(log, "task init start");
        // 转换函数
        List<TransformerExecution> transformerExecutionList = null;
        if (StrUtil.isNotBlank(transformer)) {
            List<TransformerConfig> transformerConfigList = JSONUtil.toList(transformer, TransformerConfig.class);
            transformerExecutionList = TransformerUtil.buildTransformerInfo(transformerConfigList);
        }

        TaskPluginCollector readerCollector = new StdoutPluginCollector(config, taskCommunication, PluginType.READER);

        // 当前模式使用非阻塞队列,
//        MemoryChannel memoryChannel = new MemoryChannel(channelConfig);
        MemoryNonBlockingChannel memoryChannel = new MemoryNonBlockingChannel(channelConfig);
        memoryChannel.setCommunication(taskCommunication);
        recordReceiver = new RecordExchanger(taskId, memoryChannel, taskCommunication, transformerExecutionList, readerCollector);

        // 根据类型创建对应writer
        // writer 配置
        writer = WriterProcessor.getWriterName(writerName);
        writer.setConfig(writerConfig);
        TaskPluginCollector writerCollector = new StdoutPluginCollector(config, taskCommunication, PluginType.WRITER);
        writer.setTaskPluginCollector(writerCollector);
        writer.init();

        // 根据对应类型创建对应reader
        // reader 配置
        reader = ReaderProcessor.getReaderName(readerName);
        reader.setConfig(readerConfig);
        reader.setTaskPluginCollector(readerCollector);
        reader.init();

        writer.prepare();
        reader.prepare();
        LogUtils.info(log, "task init end， reader: {}, writer:{}, transformer function:{}", reader, writer, transformerExecutionList);
        this.startTimeStamp = System.currentTimeMillis();
        lastTaskGroupContainerCommunication = new Communication();
    }

    public void start() {
        LogUtils.warn(log, "nothing");
    }

    public void sendData(String data) {

        try {
            long start = System.currentTimeMillis();
            reader.startReader(recordReceiver, data);
            long readerEnd = System.currentTimeMillis();
            writer.startWrite(recordReceiver);
            long end = System.currentTimeMillis();
            log.info("reader: {} ms, writer：{} ms", readerEnd - start, end - readerEnd);
            lastTaskGroupContainerCommunication = reportTaskGroupCommunication(lastTaskGroupContainerCommunication, 1);
        } catch (Exception e) {
            String message = String.format("读写数据异常,reader:%s, writer: %s, data: %s, error: %s", readerName, writerName, data, e.getMessage());
            LogUtils.error(log, message, e);
            throw new IntegrationValidException(message, e);
        }
    }

    @Override
    public void stop() {
        recordReceiver.shutdown();
        reader.destroy();
        writer.destroy();
    }

    @Override
    public void shutdown() {
        recordReceiver.shutdown();
    }


    private Communication reportTaskGroupCommunication(Communication lastTaskGroupContainerCommunication, int taskCount) {
        long currentTimeStamp = System.currentTimeMillis();
        long totalCosts = (currentTimeStamp - this.startTimeStamp) / 1000;
        if (totalCosts == 0L) {
            totalCosts = 1L;
        }

        taskCommunication.setTimestamp(System.currentTimeMillis());
        Communication reportCommunication = CommunicationTool.getReportCommunication(taskCommunication,
                lastTaskGroupContainerCommunication, taskCount);

        // 字节速率
        long byteSpeedPerSecond = taskCommunication.getLongCounter(CommunicationTool.READ_SUCCEED_BYTES)
                / totalCosts;

        long recordSpeedPerSecond = taskCommunication.getLongCounter(CommunicationTool.READ_SUCCEED_RECORDS)
                / totalCosts;

        lastTaskGroupContainerCommunication.setLongCounter(CommunicationTool.BYTE_SPEED, byteSpeedPerSecond);
        lastTaskGroupContainerCommunication.setLongCounter(CommunicationTool.RECORD_SPEED, recordSpeedPerSecond);

        LogUtils.info(log, "taskId: " + taskId + ", " + CommunicationTool.Stringify.getSnapshot(taskCommunication));
        return reportCommunication;
    }

}
